package com.gzyj.flink.gps.map.area;

import cn.hutool.core.lang.Pair;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.alibaba.fastjson.support.spring.FastJsonRedisSerializer;
import com.gzyj.flink.driver.Driverlogininfo;
import com.gzyj.flink.dto.alarm.Kafka_VehicleAreaRiskAlarm;
import com.gzyj.flink.dto.alarm.RiskData;
import com.gzyj.flink.gps.GPSPosition;
import com.gzyj.flink.gps.config.RedisKeyPrefixConfig;
import com.gzyj.flink.gps.factory.MySQLMybatisSessionFactory;
import com.gzyj.flink.gps.factory.RedisSessionFactory;
import com.gzyj.flink.gps.mapper.mysql.MapVehareaMapper;
import com.gzyj.flink.gps.mapper.mysql.MapareaMapper;
import com.gzyj.flink.gps.service.AreaService;
import com.gzyj.flink.gps.service.YjGpsAddrService;
import com.gzyj.flink.gps.utils.TimeUtils;
import com.gzyj.flink.pojo.mysql.set.Maparea;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.ibatis.session.SqlSession;
import org.apache.ibatis.session.SqlSessionFactory;
import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.StringRedisSerializer;

import java.util.concurrent.TimeUnit;

public class AreaRiskAlarmDataProcess extends RichMapFunction<GPSPosition, Kafka_VehicleAreaRiskAlarm> {

    LettuceConnectionFactory redisFactory;

    SqlSessionFactory mysql;

    private AreaService areaService;

    private YjGpsAddrService yjGpsAddrService;

    private RedisTemplate<String, Object> getRedisTemplate(LettuceConnectionFactory redisSessionFactory) {
        RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>();
        redisTemplate.setConnectionFactory(redisSessionFactory);
        redisTemplate.setKeySerializer(new StringRedisSerializer());
        redisTemplate.setValueSerializer(new FastJsonRedisSerializer<>(Object.class));
        redisTemplate.setHashKeySerializer(new StringRedisSerializer());
        redisTemplate.setHashValueSerializer(new FastJsonRedisSerializer<>(Object.class));
        redisTemplate.afterPropertiesSet();
        return redisTemplate;
    }


    private Kafka_VehicleAreaRiskAlarm SendRiskAlarm(RiskData data, GPSPosition gps, Driverlogininfo driverLoginData) {

        Kafka_VehicleAreaRiskAlarm alarm = new Kafka_VehicleAreaRiskAlarm();
        alarm.setVehicleId(gps.getVehId());
        alarm.setNote("风险管控报警");
        alarm.setInRiskTime(data.getInRiskTime());
        alarm.setRiskId(data.getRiskId());
        alarm.setRiskName(data.getRiskName());
        alarm.setRiskLevel(data.getRiskLevel());
        alarm.setInRiskLongitude(data.getInRiskLongitude());
        alarm.setInRiskLatitude(data.getInRiskLatitude());
        alarm.setInRiskRoadName(yjGpsAddrService.getAddress(data.getInRiskLatitude(), data.getInRiskLongitude()));
        alarm.setOutRiskTime(data.getOutRiskTime());
        alarm.setOutRiskLongitude(data.getOutRiskLongitude());
        alarm.setOutRiskLatitude(data.getOutRiskLatitude());
        alarm.setOutRiskRoadName(yjGpsAddrService.getAddress(data.getOutRiskLatitude(), data.getOutRiskLongitude()));
        alarm.setInDependencyTime(data.getInDependencyTime());
        alarm.setDependencyId(data.getDependencyId());
        alarm.setDependencyName(data.getDependencyName());
        alarm.setInDependencyLongitude(data.getInDependencyLongitude());
        alarm.setInDependencyLatitude(data.getInDependencyLatitude());
        alarm.setInDependencyRoadName(yjGpsAddrService.getAddress(data.getInDependencyLatitude(), data.getInDependencyLongitude()));
        alarm.setDriver(driverLoginData.getDriverName());
        alarm.setDriverNo(driverLoginData.getDriverNo());
        alarm.setQualificationCode(driverLoginData.getDriverNo());
        return alarm;


    }


    @Override
    public void open(Configuration parameters) throws Exception {
        redisFactory = RedisSessionFactory.getRedisFactory();
        mysql = MySQLMybatisSessionFactory.getSqlSessionFactory();
        areaService = new AreaService();
        yjGpsAddrService = new YjGpsAddrService();
    }

    public Driverlogininfo TryGetData_DLD(String vehId, RedisTemplate<String, Object> clusterRedisTemplate) {
        Object obj = clusterRedisTemplate.opsForValue().get(RedisKeyPrefixConfig.RedisKey_DriverSignOn + vehId);
        if (obj == null) {
            return new Driverlogininfo();
        } else {
            return JSON.parseObject(JSON.toJSONString(obj, SerializerFeature.DisableCircularReferenceDetect), Driverlogininfo.class);
        }

    }

    @Override
    public Kafka_VehicleAreaRiskAlarm map(GPSPosition gpsPosition) throws Exception {
        RedisTemplate<String, Object> clusterRedisTemplate = getRedisTemplate(redisFactory);
        SqlSession sqlSession = mysql.openSession(true);

        Kafka_VehicleAreaRiskAlarm areaRiskAlarm = new Kafka_VehicleAreaRiskAlarm();
        try {

            MapVehareaMapper mapVehareaMapper = sqlSession.getMapper(MapVehareaMapper.class);
            MapareaMapper mapareaMapper = sqlSession.getMapper(MapareaMapper.class);

            Driverlogininfo driverLoginData = TryGetData_DLD(gpsPosition.getVehId(), clusterRedisTemplate);
            // 获取运行数据
            Object obj = clusterRedisTemplate.opsForValue().get(RedisKeyPrefixConfig.RedisKey_RiskArea + gpsPosition.getVehId());
            RiskData riskData;
            if (obj == null) {
                riskData = new RiskData();
            } else {
                riskData = JSON.parseObject(JSON.toJSONString(obj, SerializerFeature.DisableCircularReferenceDetect), RiskData.class);
            }

            Pair<Boolean, Integer> p = TimeUtils.CheckData(riskData.getGpsTime(), gpsPosition.getTIME());
            if (!p.getKey()) {
                return null;
            } else {
                riskData.GpsTime = gpsPosition.getGPSTimeExt();
            }
            boolean isInRisk = riskData.InRisk;
            boolean isInDependency = riskData.InDependency;
            Pair<Boolean, Maparea> p1 = areaService.InRisk(gpsPosition.getVehId(), gpsPosition.getLngExt(), gpsPosition.getLatExt(), false, mapVehareaMapper, mapareaMapper);
            Maparea mySetting = p1.getValue();
            if (p1.getKey()) {
                //在风险区
                if (!isInRisk) {
                    //进入了风险区,且之前没在风险区
                    riskData.InRisk = true;
                    riskData.Status = 1;
                    riskData.InRiskTime = gpsPosition.getGPSTimeExt();
                    riskData.RiskId = mySetting.getAreaid();
                    riskData.RiskName = mySetting.getName();
                    riskData.RiskLevel = mySetting.getEpidemicrisk();
                    riskData.InRiskLongitude = gpsPosition.getLngExt();
                    riskData.InRiskLatitude = gpsPosition.getLatExt();
                    riskData.OutRiskTime = TimeUtils.MIN_VALUE();
                    riskData.OutRiskLongitude = 0;
                    riskData.OutRiskLatitude = 0;

                    riskData.InDependency = false;
                    riskData.InDependencyTime = gpsPosition.getGPSTimeExt();
                    riskData.DependencyId = -1;
                    riskData.DependencyName = "";
                    riskData.InDependencyLongitude = 0;
                    riskData.InDependencyLatitude = 0;

                    //与InDependency互斥
                    riskData.InDependency = false;

                } else {
                    //进入了风险区,且之前也在风险区
                }
            } else {

                //不在风险区
                if (isInRisk) {
                    //不在风险区,且之前在风险区，记录出风险区的信息
                    riskData.OutRiskTime = gpsPosition.getGPSTimeExt();
                    riskData.OutRiskLongitude = gpsPosition.getLngExt();
                    riskData.OutRiskLatitude = gpsPosition.getLatExt();
                    riskData.InRisk = false;
                } else {
                    //不在风险区,且之前也不在风险区
                }

                //如果曾经进过风险区.则判断是否在属地
                if (riskData.Status == 1) {
                    Pair<Boolean, Maparea> p2 = areaService.InDependency(gpsPosition.getVehId(), gpsPosition.getLngExt(), gpsPosition.getLatExt(), false, mapVehareaMapper, mapareaMapper);
                    Maparea dependencySetting = p2.getValue();
                    if (p2.getKey()) {
                        //在属地
                        if (!isInDependency) {
                            //进入了属地,且之前没在属地
                            riskData.InDependency = true;
                            riskData.InDependencyTime = gpsPosition.getGPSTimeExt();
                            riskData.DependencyId = dependencySetting.getAreaid();
                            riskData.DependencyName = dependencySetting.getName();
                            riskData.InDependencyLongitude = gpsPosition.getLngExt();
                            riskData.InDependencyLatitude = gpsPosition.getLatExt();

                            //与InRisk互斥
                            riskData.InRisk = false;
                            riskData.Status = 0;

                            //存
                            clusterRedisTemplate.opsForValue().set(RedisKeyPrefixConfig.RedisKey_RiskArea + gpsPosition.getVehId(), riskData, 8, TimeUnit.HOURS);

                            areaRiskAlarm = SendRiskAlarm(riskData, gpsPosition, driverLoginData);

                        } else {
                            //进入了属地,且之前也在属地
                        }
                    }
                }
            }

            clusterRedisTemplate.opsForValue().set(RedisKeyPrefixConfig.RedisKey_RiskArea + gpsPosition.getVehId(), riskData, 8, TimeUnit.HOURS);

        } catch (Exception e) {
            e.printStackTrace();
            sqlSession.rollback();
        } finally {
            sqlSession.close();
            return areaRiskAlarm;
        }

    }
}
